package org.opens.lambda.future;

import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.CompletableFuture;

public class CompletableAction2 {

    public static class Handle {

        public static void main(String[] args) throws InterruptedException {
            CompletableFuture.supplyAsync(() -> 1)
                    //handle的参数是一个BiFunction, 第一个参数是上一步异步任务的返回值, 第二个参数是可能发生的异常
                    .handle((v, t) -> {
                        try {
                            Thread.sleep(2000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        return v + 10;
                    })
                    .whenComplete((v, t) -> System.out.println(v))
                    //如果想在所有操作执行完毕之后执行一个操作, 可以链式调用thenRun
                    //注意的是, 这种方式是创建一个Runnable接口的实现传入, 并没有接受上一步的结果
                    //thenRunAsync是一种异步处理方式
                    .thenRun(() -> System.out.println("异步结束了"));
            //CompletableFuture.supplyAsync创建的线程默认为守护线程, 可以自定义非守护线程线程池并传入达到创建非守护线程执行异步任务的效果
            //如果想验证CompletableFuture.supplyAsync创建的线程默认为守护线程, 只需要把线面的主线程睡眠时间调整的小于异步任务的沉睡时间观察输出即可
            Thread.sleep(3000);
            System.out.println("主线程结束了");
        }

    }

    public static class ThenAccept {

        public static void main(String[] args) throws InterruptedException {
            CompletableFuture.supplyAsync(() -> 1)
                    //处理结果集
                    .thenApply(v -> v + 10)
                    //thenAccept接受的是一个Consumer, 所以相当于是一个终端操作.
                    .thenAccept(System.out::println);
            Thread.sleep(3000);
        }

    }

    public static class ThenCompose {

        public static void main(String[] args) throws InterruptedException {
            CompletableFuture.supplyAsync(() -> 1)
                    //这种方式是将前一步CompletableFuture.supplyAsync执行的结果传递给新的CompletableFuture.supplyAsync去执行.
                    .thenCompose(i -> CompletableFuture.supplyAsync(() -> 10 * i))
                    .thenAccept(System.out::println);
            Thread.sleep(3000);
        }

    }

    /**
     * 输出:
     * r1:1
     * r2:2.12
     * 3.12
     */
    public static class ThenCombine {

        public static void main(String[] args) throws InterruptedException {
            CompletableFuture.supplyAsync(() -> 1)
                    /*
                    1. thenCombine方法的第一个参数还是一个CompletableFuture对象, 而第二个参数是一个BiFunction, 从输出结果中可以看出
                        - thenCombine方法的第二个参数的第一个参数是外界的CompletableFuture对象的输出结果;
                        - thenCombine方法的第二个参数的第二个参数是内部的CompletableFuture对象的输出结果;
                     */
                    .thenCombine(CompletableFuture.supplyAsync(() -> 2.12D), (r1, r2) -> {
                        System.out.println("r1:" + r1);
                        System.out.println("r2:" + r2);
                        return r1 + r2;
                    }).thenAccept(System.out::println);
            Thread.sleep(3000);
        }

    }

    public static class ThenAcceptBoth {

        public static void main(String[] args) throws InterruptedException {
            CompletableFuture.supplyAsync(() -> 1)
                    /*
                    1. 可以发现, thenAcceptBoth的作用和thenCombine的作用是极为相似的, 只是第二个参数变成了BiConsumer, 没有了
                    返回值.
                     */
                    .thenAcceptBoth(CompletableFuture.supplyAsync(() -> 2.12D), (r1, r2) -> {
                        System.out.println("r1:" + r1);
                        System.out.println("r2:" + r2);
                        System.out.println(r1 + r2);
                    });
            Thread.sleep(3000);
        }

    }

    public static class RunAfterBoth {
        /**
         * 说明:
         * 1. runAfterBoth的第二个参数是一个Runnable接口的lambda实现, 最终会别一个线程去执行, 但是这个线程会等待两个
         * CompletableFuture对象执行完毕.
         */
        public static void main(String[] args) throws InterruptedException {
            CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(1500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("第一个CompletableFuture执行完了");
                return 1;
            }).runAfterBoth(CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("第二个CompletableFuture执行完了");
                return 2;
            }), () -> System.out.println("done"));
            Thread.sleep(3000);
        }

    }

    /**
     * 说明:
     * 1. applyToEither方法的执行原则是:
     * - 两个CompletableFuture对象谁先执行完, 谁就被applyToEither的第二个参数的lambda实现调用并返回.
     * 2. 这种方式可以应用到这样的一种结构中:
     * - 一个方法可以从两台服务器上请求到相同的数据, 就同时去尝试请求这两个服务器, 只要其中一个做出了响应, 就立即
     * 返回结果, 屏蔽另一个的返回值.
     */
    public static class ApplyToEither {

        public static void main(String[] args) throws InterruptedException {
            CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("I'm 1");
                return 1;
            }).applyToEither(CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("I'm 2");
                return 2;
            }), result -> result).thenAccept(x -> {
                System.out.println("最终结果为: " + x);
            });
            Thread.sleep(2000);
        }
    }

    /**
     * 说明:
     * 1. acceptEither相当于是直接进行消费, applyToEither的第二个参数是Function, 而acceptEither的第二个参数是Consumer.
     */
    public static class AcceptEither {

        public static void main(String[] args) throws InterruptedException {
            CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("I'm 1");
                return 1;
            }).acceptEither(CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("I'm 2");
                return 2;
            }), System.out::println);
            Thread.sleep(2000);
        }

    }

    public static class AllOf {

        /**
         * 输出:
         * 做出返回: 300
         * 做出返回: 400
         * 做出返回: 200
         * 做出返回: 100
         * done
         * 说明:
         * 1. allOf的参数是一个CompletableFuture[], 其功能是等待CompletableFuture[]中的每一个任务都执行完毕之后才会
         * 执行该函数的下一步.
         * - 并不会使主线程阻塞.
         */
        public static void main(String[] args) throws InterruptedException {
            Random random = new Random();
            CompletableFuture[] completableFutures = Arrays.asList(1, 2, 3, 4).stream()
                    .map(x -> CompletableFuture.supplyAsync(() -> {
                        try {
                            Thread.sleep(random.nextInt(3000));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("做出返回: " + x * 100);
                        return x * 100;
                    }))
                    .toArray(CompletableFuture[]::new);
            CompletableFuture.allOf(completableFutures)
                    .thenRun(() -> System.out.println("done"));
            //使主线程等待
            Thread.currentThread().join();
        }
    }

    public static class AnyOf {

        /**
         * 输出:
         * 做出返回: 300
         * done
         * 做出返回: 100
         * 做出返回: 200
         * 做出返回: 400
         * 说明:
         * 1. anyof表示只要 CompletableFuture[]参数中的一个任务执行完毕就执行下一步.
         */
        public static void main(String[] args) throws InterruptedException {
            Random random = new Random();
            CompletableFuture[] completableFutures = Arrays.asList(1, 2, 3, 4).stream()
                    .map(x -> CompletableFuture.supplyAsync(() -> {
                        try {
                            Thread.sleep(random.nextInt(3000));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("做出返回: " + x * 100);
                        return x * 100;
                    }))
                    .toArray(CompletableFuture[]::new);
            CompletableFuture.anyOf(completableFutures)
                    .thenRun(() -> System.out.println("done"));
            //使主线程等待
            Thread.currentThread().join();
        }

    }

}
